package lsm

import (
	"errors"
	"learning/gooop/etcd/raft/config"
	"learning/gooop/etcd/raft/debug"
	"learning/gooop/etcd/raft/model"
	"learning/gooop/etcd/raft/roles"
	"learning/gooop/etcd/raft/rpc"
	"learning/gooop/etcd/raft/store"
	"learning/gooop/etcd/raft/timeout"
	"sync"
	"time"
)

// tLeaderState presents a leader node
type tLeaderState struct {
	model.TEventDrivenModel

	context    iRaftStateContext
	mInitOnce  sync.Once
	mStartOnce sync.Once

	// update: leInit / leLeaderHeartbeat
	mTerm int64

	// update: leInit / leDisposing
	mDisposedFlag bool

	// update: leVoteToCandidate
	mVotedTerm        int64
	mVotedCandidateID string
	mVotedTimestamp   int64
}

// trigger: init()
// args: empty
const leInit = "leader.init"

// trigger: Start()
// args: empty
const leStart = "leader.Start"

// trigger: whenNewLeaderAnnouncedThenSwitchToFollower
// args: empty
const leDiposing = "leader.Disposing"

// trigger : Heartbeat() / AppendLog()
// args: term int64
const leNewLeaderAnnounced = "leader.NewLeaderAnnounced"

// trigger: RequestVote()
// args: *rpc.RequestVoteCmd
const leBeforeRequestVote = "leader.BeforeRequestVote"

// trigger:
// args: *rpc.RequestVoteCmd
const leVoteToCandidate = "leader.VoteToCandidate"

// trigger: handleHeartbeat()
// args: term int64
const leHeartbeatRejected = "leader.HeartbeatRejected"

func newLeaderState(ctx iRaftStateContext, term int64) IRaftState {
	it := new(tLeaderState)
	it.init(ctx, term)
	return it
}

func (me *tLeaderState) init(ctx iRaftStateContext, term int64) {
	me.mInitOnce.Do(func() {
		me.context = ctx
		me.mTerm = term

		me.initEventHandlers()
		me.Raise(leInit)

		debug.LeaderNodeID = ctx.Config().GetID()
	})
}

func (me *tLeaderState) initEventHandlers() {
	// write only logic
	me.hookEventsForDisposedFlag()
	me.hookEventsForVotedTerm()

	// read only logic
	me.Hook(leStart,
		me.whenStartThenBeginHeartbeatToOthers)
	me.Hook(leNewLeaderAnnounced,
		me.whenNewLeaderAnnouncedThenSwitchToFollower)
	me.Hook(leHeartbeatRejected,
		me.whenHeartbeatRejectedThenSwitchToFollower)
}

func (me *tLeaderState) hookEventsForDisposedFlag() {
	me.Hook(leInit, func(e string, args ...interface{}) {
		me.mDisposedFlag = false
	})

	me.Hook(leDiposing, func(e string, args ...interface{}) {
		me.mDisposedFlag = true
	})
}

func (me *tLeaderState) hookEventsForVotedTerm() {
	me.Hook(leBeforeRequestVote, func(e string, args ...interface{}) {
		// check last vote timeout
		if me.mVotedTerm == 0 {
			return
		}

		if time.Duration(time.Now().UnixNano()-me.mVotedTimestamp)*time.Nanosecond >= timeout.ElectionTimeout {
			me.mVotedTerm = 0
			me.mVotedTimestamp = 0
			me.mVotedCandidateID = ""
		}
	})

	me.Hook(leVoteToCandidate, func(e string, args ...interface{}) {
		// after vote to candidate
		cmd := args[0].(*rpc.RequestVoteCmd)
		me.mVotedTerm = cmd.Term
		me.mVotedCandidateID = cmd.CandidateID
		me.mVotedTimestamp = time.Now().UnixNano()
	})
}

func (me *tLeaderState) Heartbeat(cmd *rpc.HeartbeatCmd, ret *rpc.HeartbeatRet) error {
	// check term
	if cmd.Term <= me.mTerm {
		ret.Code = rpc.HBTermMismatch
		return nil
	}

	// new leader
	me.Raise(leNewLeaderAnnounced, cmd.Term)

	// return ok
	ret.Code = rpc.HBOk
	return nil
}

func (me *tLeaderState) AppendLog(cmd *rpc.AppendLogCmd, ret *rpc.AppendLogRet) error {
	// check term
	if cmd.Term <= me.mTerm {
		ret.Code = rpc.ALTermMismatch
		return nil
	}

	// new leader
	me.Raise(leNewLeaderAnnounced, cmd.Term)

	// return ok
	ret.Code = rpc.ALInternalError
	return nil
}

func (me *tLeaderState) CommitLog(cmd *rpc.CommitLogCmd, ret *rpc.CommitLogRet) error {
	// just ignore
	ret.Code = rpc.CLInternalError
	return nil
}

func (me *tLeaderState) RequestVote(cmd *rpc.RequestVoteCmd, ret *rpc.RequestVoteRet) error {
	me.Raise(leBeforeRequestVote, cmd)

	// check voted term
	if cmd.Term < me.mVotedTerm {
		ret.Code = rpc.RVTermMismatch
		return nil
	}

	if cmd.Term == me.mVotedTerm {
		if me.mVotedCandidateID != "" && me.mVotedCandidateID != cmd.CandidateID {
			// already vote another
			ret.Code = rpc.RVVotedAnother
			return nil
		} else {
			// already voted
			ret.Code = rpc.RVOk
			return nil
		}
	}

	if cmd.Term > me.mVotedTerm {
		// new term, check log
		if cmd.LastLogIndex >= me.context.Store().LastCommittedIndex() {
			// good log
			me.Raise(leVoteToCandidate, cmd)
			ret.Code = rpc.RVOk

		} else {
			// bad log
			ret.Code = rpc.RVLogMismatch
		}

		return nil
	}

	// should not reach here
	ret.Code = rpc.RVTermMismatch
	return nil
}

func (me *tLeaderState) Role() roles.RaftRole {
	return roles.Leader
}

func (me *tLeaderState) Start() {
	me.mStartOnce.Do(func() {
		me.Raise(leStart)
	})
}

func (me *tLeaderState) whenStartThenBeginHeartbeatToOthers(_ string, _ ...interface{}) {
	go func() {
		for !me.mDisposedFlag && debug.KilledNodeID != me.context.Config().GetID() {
			_ = me.boardcast(func(_ config.IRaftNodeConfig, client rpc.IRaftRPC) error {
				return me.handleHeartbeat(client)
			})
			time.Sleep(timeout.HeartbeatInterval)
		}
	}()
}

func (me *tLeaderState) boardcast(action func(config.IRaftNodeConfig, rpc.IRaftRPC) error) error {
	for _, it := range me.context.Config().GetNodes() {
		if it.GetID() == me.context.Config().GetID() {
			continue
		}

		e := me.context.RaftClientService().UsingRaft(it.GetID(), func(client rpc.IRaftRPC) error {
			return action(it, client)
		})
		if e != nil {
			return e
		}
	}

	return nil
}

func (me *tLeaderState) handleHeartbeat(client rpc.IRaftRPC) error {
	cmd := new(rpc.HeartbeatCmd)
	cmd.Term = me.mTerm
	cmd.LeaderID = me.context.Config().GetID()

	ret := new(rpc.HeartbeatRet)
	e := client.Heartbeat(cmd, ret)
	if e != nil {
		return e
	}

	switch ret.Code {
	case rpc.HBTermMismatch:
		me.Raise(leHeartbeatRejected, ret.Term)
		break
	}

	return nil
}

func (me *tLeaderState) whenNewLeaderAnnouncedThenSwitchToFollower(_ string, args ...interface{}) {
	me.Raise(leDiposing)

	term := args[0].(int64)
	me.context.HandleStateChanged(newFollowerState(me.context, term))
}

func (me *tLeaderState) whenHeartbeatRejectedThenSwitchToFollower(_ string, args ...interface{}) {
	me.Raise(leDiposing)

	term := args[0].(int64)
	me.context.HandleStateChanged(newFollowerState(me.context, term))
}

func (me *tLeaderState) ExecuteKVCmd(cmd *rpc.KVCmd, ret *rpc.KVRet) error {
	switch cmd.OPCode {
	case rpc.KVGet:
		return me.handleKVGet(cmd, ret)

	case rpc.KVPut:
		return me.handleKVPut(cmd, ret)

	case rpc.KVDel:
		return me.handleKVDel(cmd, ret)
	}

	return nil
}

func (me *tLeaderState) handleKVGet(cmd *rpc.KVCmd, ret *rpc.KVRet) error {
	e, v := me.context.Store().Get(cmd.Key)
	if e != nil {
		ret.Code = rpc.KVInternalError
		return e
	}

	ret.Code = rpc.KVOk
	ret.Content = v
	return nil
}

func (me *tLeaderState) handleKVPut(cmd *rpc.KVCmd, ret *rpc.KVRet) error {
	kvcmd := new(store.PutCmd)
	kvcmd.Key = cmd.Key
	kvcmd.Value = cmd.Content

	// create/append/commit log
	e := me.broadcastKVCmd(kvcmd, ret)
	if e != nil {
		return e
	}

	// apply cmd
	return me.context.Store().Put(cmd.Key, cmd.Content)
}

func (me *tLeaderState) handleKVDel(cmd *rpc.KVCmd, ret *rpc.KVRet) error {
	kvcmd := new(store.DelCmd)
	kvcmd.Key = cmd.Key

	// create/append/commit log
	e := me.broadcastKVCmd(kvcmd, ret)
	if e != nil {
		return e
	}

	// apply cmd
	return me.context.Store().Put(cmd.Key, cmd.Content)
}

func (me *tLeaderState) broadcastKVCmd(cmd store.IKVCmd, ret *rpc.KVRet) error {
	// create log
	st := me.context.Store()
	log := new(model.LogEntry)
	log.Term = me.mTerm
	log.Tag = cmd.Tag()
	log.Index = st.LastCommittedIndex() + 1
	log.PrevTerm = st.LastCommittedTerm()
	log.PrevIndex = st.LastCommittedIndex()
	log.Command = cmd.Marshal()

	// append log
	e := st.Append(log)
	if e != nil {
		ret.Code = rpc.KVInternalError
		return e
	}

	// ask other nodes to append log
	alcmd := new(rpc.AppendLogCmd)
	alcmd.Term = me.mTerm
	alcmd.LeaderID = me.context.Config().GetID()
	alcmd.Entry = log

	sumOk := []int{0}
	_ = me.boardcast(func(_ config.IRaftNodeConfig, client rpc.IRaftRPC) error {
		r := new(rpc.AppendLogRet)
		e := client.AppendLog(alcmd, r)
		if e != nil {
			return e
		}

		switch r.Code {
		case rpc.ALOk:
			sumOk[0]++
			break

		case rpc.ALTermMismatch:
			// todo: fixme
			break

		case rpc.ALIndexMismatch:
			// todo: fixme
			break
		}

		return nil
	})

	// wait for most nodes
	if sumOk[0] >= len(me.context.Config().GetNodes())/2 {
		// commit log
		clcmd := new(rpc.CommitLogCmd)
		clcmd.LeaderID = me.context.Config().GetID()
		clcmd.Term = me.mTerm
		clcmd.Index = log.Index

		_ = me.boardcast(func(_ config.IRaftNodeConfig, client rpc.IRaftRPC) error {
			r := new(rpc.CommitLogRet)
			e := client.CommitLog(clcmd, r)
			if e != nil {
				return e
			}

			switch r.Code {
			case rpc.CLInternalError:
				// todo: fixme
				break

			case rpc.CLLogNotFound:
				// todo: fixme
				break

			case rpc.CLOk:
				return nil
			}

			return nil
		})

		// ok
		return nil

	} else {
		return gErrorCannotReachAgreement
	}
}

var gErrorCannotReachAgreement = errors.New("cannot reach agreement")
